package com.example.mq.server.virtualhost;

import com.example.mq.common.BinaryTool;
import com.example.mq.common.MqException;
import com.example.mq.common.Response;
import com.example.mq.common.payloads.PayloadResponseSubscribe;
import com.example.mq.server.MemoryData;
import com.example.mq.server.ServerAPI;
import com.example.mq.server.consumermanager.ConsumerManager;
import com.example.mq.server.diskdatamanager.DiskDataCenter;
import com.example.mq.server.messagemanager.BasicProperties;
import com.example.mq.server.messagemanager.Message;

import java.io.DataOutputStream;
import java.io.IOException;
import java.net.Socket;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

public class VirtualHost implements ServerAPI {
    private String virtualHostName;

    private MemoryData memoryData = new MemoryData();

    private DiskDataCenter diskDataCenter = new DiskDataCenter();

    private Router router = new Router();

    private ConsumerManager consumerManager = new ConsumerManager(this);

    // 引入一个顺序表，存储已连接的客户端
    private ConcurrentHashMap<String, Socket> sessions = new ConcurrentHashMap<String, Socket>();

    // 操作交换机的锁对象
    private final Object exchangeLocker = new Object();
    // 操作队列的锁对象
    private final Object queueLocker = new Object();


    public VirtualHost(String virtualHostName) {
        this.virtualHostName = virtualHostName;

        diskDataCenter.init();

        // 从硬盘中加载内存
        try {
            memoryData.recovery(diskDataCenter);
        } catch (IOException | MqException | ClassNotFoundException e) {
            e.printStackTrace();
            System.out.println("[VirtualHost] 加载内存失败");
        }
    }

    public MemoryData getMemoryData() {
        return memoryData;
    }

    public void setSessions(ConcurrentHashMap<String, Socket> sessions) {
        this.sessions = sessions;
    }

    @Override
    public boolean exchangeDeclare(String exchangeName, ExchangeType exchangeType) {
        // 拿到交换机名字
        exchangeName = virtualHostName + exchangeName;
        try {
            synchronized (exchangeLocker) {

                // 1. 判定该交换机是否已经存在
                Exchange existsExchange = memoryData.getExchange(exchangeName);
                if (existsExchange != null) {
                    // 该交换机已经存在!
                    System.out.println("[VirtualHost] 交换机已经存在! exchangeName=" + exchangeName);
                    return true;
                }

                // 2. 创建交换机
                Exchange exchange = new Exchange();
                exchange.setName(exchangeName);
                exchange.setType(exchangeType);

                // 3.写入硬盘
                diskDataCenter.insertExchange(exchange);

                // 4.写入内存
                memoryData.insertExchange(exchange);

                System.out.println("[BrokerServer] 交换机创建完成! exchangeName=" + exchangeName);
            }
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] 交换机创建失败! exchangeName=" + exchangeName);
            e.printStackTrace();
            return false;
        }
    }

    @Override
    public boolean exchangeDelete(String exchangeName) {
        // 拿到交换机名字
        exchangeName = virtualHostName + exchangeName;
        try {
            synchronized (exchangeLocker) {
                // 1. 先找到对应的交换机.
                Exchange toDelete = memoryData.getExchange(exchangeName);
                if (toDelete == null) {
                    throw new MqException("[VirtualHost] 交换机不存在无法删除!");
                }

                // 2.从硬盘上删除
                diskDataCenter.deleteExchange(exchangeName);

                // 3.从内存上删除
                memoryData.deleteExchange(exchangeName);
                System.out.println("[VirtualHost] 交换机删除成功! exchangeName=" + exchangeName);
            }
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] 交换机删除失败! exchangeName=" + exchangeName);
            e.printStackTrace();
            return false;
        }
    }

    @Override
    public boolean queueDeclare(String queueName) {
        // 拿到队列名字
        queueName = virtualHostName + queueName;
        try {
            synchronized (queueLocker) {
                // 1. 判定队列是否存在
                MQueue existsQueue = memoryData.getQueue(queueName);
                if (existsQueue != null) {
                    System.out.println("[VirtualHost] 队列已经存在! queueName=" + queueName);
                    return true;
                }
                // 2. 创建队列对象
                MQueue queue = new MQueue();
                queue.setName(queueName);

                // 3.写入硬盘
                diskDataCenter.insertQueue(queue);

                // 4.写入内存
                memoryData.insertQueue(queue);
                System.out.println("[VirtualHost] 队列创建成功! queueName=" + queueName);
            }
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] 队列创建失败! queueName=" + queueName);
            e.printStackTrace();
            return false;
        }
    }

    @Override
    public boolean queueDelete(String queueName) {
        // 拿到队列名字
        queueName = virtualHostName + queueName;
        try {
            synchronized (queueLocker) {
                // 1. 根据队列名字, 查询下当前的队列对象
                MQueue queue = memoryData.getQueue(queueName);
                if (queue == null) {
                    throw new MqException("[VirtualHost] 队列不存在! 无法删除! queueName=" + queueName);
                }

                // 2.从硬盘中删除
                diskDataCenter.deleteQueue(queueName);

                // 3.从内存中删除
                memoryData.deleteQueue(queueName);
                System.out.println("[VirtualHost] 删除队列成功! queueName=" + queueName);
            }
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] 删除队列失败! queueName=" + queueName);
            e.printStackTrace();
            return false;
        }
    }

    @Override
    public boolean bind(String queueName, String exchangeName, String bindingKey) {

        queueName = virtualHostName + queueName;
        exchangeName = virtualHostName + exchangeName;

        try {
            synchronized (exchangeLocker) {
                synchronized (queueLocker) {

                    // 1. 判定当前的绑定是否已经存在了.
                    Binding existsBinding = memoryData.getBinding(exchangeName, queueName);
                    if (existsBinding != null) {
                        throw new MqException("[VirtualHost] binding 已经存在! queueName=" + queueName
                                + ", exchangeName=" + exchangeName);
                    }

                    // 2. 验证 bindingKey 是否合法.
                    if (!router.checkBindingKey(bindingKey)) {
                        throw new MqException("[VirtualHost] bindingKey 非法! bindingKey=" + bindingKey);
                    }

                    // 3. 创建 Binding 对象
                    Binding binding = new Binding();
                    binding.setExchangeName(exchangeName);
                    binding.setQueueName(queueName);
                    binding.setBindingKey(bindingKey);

                    // 4. 检查交换机和队列是否存在
                    MQueue queue = memoryData.getQueue(queueName);
                    if (queue == null) {
                        throw new MqException("[VirtualHost] 队列不存在! queueName=" + queueName);
                    }
                    Exchange exchange = memoryData.getExchange(exchangeName);
                    if (exchange == null) {
                        throw new MqException("[VirtualHost] 交换机不存在! exchangeName=" + exchangeName);
                    }

                    // 5.写入硬盘
                    diskDataCenter.insertBinding(binding);

                    // 6.写入内存
                    memoryData.insertBinding(binding);
                }
            }
            System.out.println("[VirtualHost] 绑定创建成功! exchangeName=" + exchangeName
                    + ", queueName=" + queueName);
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] 绑定创建失败! exchangeName=" + exchangeName
                    + ", queueName=" + queueName);
            e.printStackTrace();
            return false;
        }
    }

    @Override
    public boolean unbind(String queueName, String exchangeName) {
        queueName = virtualHostName + queueName;
        exchangeName = virtualHostName + exchangeName;
        try {
            synchronized (exchangeLocker) {
                synchronized (queueLocker) {
                    // 1. 获取 binding 看是否已经存在~
                    Binding binding = memoryData.getBinding(exchangeName, queueName);
                    if (binding == null) {
                        throw new MqException("[VirtualHost] 删除绑定失败! 绑定不存在! exchangeName=" + exchangeName + ", queueName=" + queueName);
                    }

                    // 2.从硬盘中删除
                    diskDataCenter.deleteBinding(binding);

                    // 3.从内存中删除
                    memoryData.deleteBinding(binding);
                    System.out.println("[VirtualHost] 删除绑定成功!");
                }
            }
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] 删除绑定失败!");
            e.printStackTrace();
            return false;
        }
    }

    @Override
    public boolean produceMessage(String exchangeName, String routingKey, byte[] body) {
        try {
            // 1. 拿到交换机的名字
            exchangeName = virtualHostName + exchangeName;

            // 2. 检查 routingKey 是否合法
            if (!router.checkRoutingKey(routingKey)) {
                throw new MqException("[VirtualHost] routingKey 非法! routingKey=" + routingKey);
            }

            // 3. 查找交换机对象
            Exchange exchange = memoryData.getExchange(exchangeName);
            if (exchange == null) {
                throw new MqException("[VirtualHost] 交换机不存在! exchangeName=" + exchangeName);
            }
            // 4. 判定交换机的类型
            if (exchange.getType() == ExchangeType.DIRECT) {
                // 按照直接交换机的方式来转发消息
                // 以 routingKey 作为队列的名字, 直接把消息写入指定的队列中.
                // 此时, 可以无视绑定关系.
                String queueName = virtualHostName + routingKey;
                // 5. 构造消息对象

                Message message = Message.createMessageWithId(routingKey, body);
                // 6. 查找该队列名对应的对象
                MQueue queue = memoryData.getQueue(queueName);
                if (queue == null) {
                    throw new MqException("[VirtualHost] 队列不存在! queueName=" + queueName);
                }
                // 7. 队列存在, 直接给队列中写入消息
                sendMessage(queue, message);
            } else {
                // 按照 fanout 和 topic 的方式来转发.
                // 5. 找到该交换机关联的所有绑定, 并遍历这些绑定对象
                ConcurrentHashMap<String, Binding> bindingsMap = memoryData.getBindings(exchangeName);
                for (Map.Entry<String, Binding> entry : bindingsMap.entrySet()) {
                    // 1) 获取到绑定对象, 判定对应的队列是否存在
                    Binding binding = entry.getValue();
                    MQueue queue = memoryData.getQueue(binding.getQueueName());
                    if (queue == null) {
                        // 此处不抛出异常了. 可能此处有多个这样的队列.
                        // 不因为一个队列的失败, 影响到其他队列的消息的传输.
                        System.out.println("[VirtualHost] basicPublish 发送消息时, 发现队列不存在! queueName=" + binding.getQueueName());
                        continue;
                    }
                    // 2) 构造消息对象
                    Message message = Message.createMessageWithId(routingKey, body);
                    // 3) 判定这个消息是否能转发给该队列.
                    //    如果是 fanout, 所有绑定的队列都要转发的.
                    //    如果是 topic, 还需要判定下, bindingKey 和 routingKey 是不是匹配.
                    if (!router.route(exchange.getType(), binding, message)) {
                        continue;
                    }
                    // 4) 真正转发消息给队列
                    sendMessage(queue, message);
                }
            }
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] 消息发送失败!");
            e.printStackTrace();
            return false;
        }
    }

    private void sendMessage(MQueue queue, Message message) throws IOException, MqException, InterruptedException {

        // 写入硬盘
        diskDataCenter.sendMessage(queue, message);

        // 写入内存
        memoryData.sendMessage(queue, message);

        // 此处还需要补充一个逻辑, 通知消费者可以消费消息了.
        consumerManager.notifyConsume(queue.getName());
    }

    @Override
    public boolean processSubscribe(String consumerTag, String queueName) {
        // 构造一个 ConsumerEnv 对象, 把这个对应的队列找到, 再把这个 Consumer 对象添加到该队列中.

        queueName = virtualHostName + queueName;
        try {
            consumerManager.addConsumer(consumerTag, queueName);
            System.out.println("[VirtualHost] basicConsume 成功! queueName=" + queueName);
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] basicConsume 失败! queueName=" + queueName);
            e.printStackTrace();
            return false;
        }
    }

    @Override
    public void consumeMessage(String clientTag, Message message, MQueue queue) throws MqException, IOException, ClassNotFoundException {
        // 先知道当前这个收到的消息, 要发给哪个客户端.
        // 此处 consumerTag 其实是 channelId. 根据 channelId 去 sessions 中查询, 就可以得到对应的
        // socket 对象了, 从而可以往里面发送数据了
        // 1. 根据 channelId 找到 socket 对象
        Socket clientSocket = sessions.get(clientTag);

        if (clientSocket == null || clientSocket.isClosed()) {
            throw new MqException("[BrokerServer] 订阅消息的客户端已经关闭!");
        }
        // 2. 构造响应数据
        PayloadResponseSubscribe payloadResponseSubscribe = new PayloadResponseSubscribe();
        payloadResponseSubscribe.setRid("1234567"); // 由于这里只有响应, 没有请求, 不需要去对应. rid 暂时不需要.
        payloadResponseSubscribe.setOk(true);
        payloadResponseSubscribe.setClientTag(clientTag);

        BasicProperties basicProperties = new BasicProperties();
        System.out.println(message.toString());

        basicProperties.setMessageId(message.getMessageId());
        basicProperties.setRoutingKey(message.getRoutingKey());

        payloadResponseSubscribe.setBasicProperties(basicProperties);
        payloadResponseSubscribe.setBody(message.getBody());
        byte[] payload = BinaryTool.toBytes(payloadResponseSubscribe);
        Response response = new Response();
        // 0xc 表示服务器给消费者客户端推送的消息数据.
        response.setType(0xc);
        // response 的 payload 就是一个 SubScribeReturns
        response.setLength(payload.length);
        response.setPayload(payload);
        // 3. 把数据写回给客户端.
        //    此处的 dataOutputStream 这个对象不能 close
        //    如果 把 dataOutputStream 关闭, 就会直接把 clientSocket 里的 outputStream 也关了.
        //    此时就无法继续往 socket 中写入后续数据了.
        DataOutputStream dataOutputStream = new DataOutputStream(clientSocket.getOutputStream());
        writeResponse(dataOutputStream, response);

        // 4. 从硬盘中删除数据
        diskDataCenter.deleteMessage(queue, message);
    }

    private void writeResponse(DataOutputStream dataOutputStream, Response response) throws IOException {
        dataOutputStream.writeInt(response.getType());
        dataOutputStream.writeInt(response.getLength());
        dataOutputStream.write(response.getPayload());
        // 这个刷新缓冲区也是重要的操作!!
        dataOutputStream.flush();
    }
}
